package org.aceor.mddal.common.zk;

import com.google.common.base.Throwables;
import com.google.common.collect.Maps;
import org.aceor.mddal.common.constants.Constants;
import org.aceor.mddal.common.zk.listener.ConfigChangeListener;
import org.aceor.mddal.common.zk.object.ConfigDTO;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.UnhandledErrorListener;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ConcurrentMap;

/**
 * Created by lxue on 16/3/19.
 */
public class DynConfigClient {
    private static final Logger logger = LoggerFactory.getLogger(DynConfigClient.class);
    private static final String CONFIG_FORMATTED = "/configs/%s/%s";
    private volatile boolean isStart = false;
    private ConcurrentMap<String, PathChildrenCache> pathChildrenCacheMap = Maps.newConcurrentMap();

    public void init() {
        if (!isStart) {
            innerRegisterListeners(ZKClient.getClient());
            isStart = true;
        }
    }

    public void registerListener(String bizName, String dataId, ConfigChangeListener listener) {
        String path = String.format(CONFIG_FORMATTED, new Object[]{bizName, dataId});
        this.doRegisterListener(path, bizName, dataId, listener);
    }

    public void removeListeners(String path) {
        ((PathChildrenCache) this.pathChildrenCacheMap.get(path)).getListenable().clear();
    }

    private final void doRegisterListener(final String path, final String bizName, final String dataId, final ConfigChangeListener listener) {
        final NodeCache cache = new NodeCache(ZKClient.getClient(), path);
        cache.getListenable().addListener(new NodeCacheListener() {
            public void nodeChanged() throws Exception {

                byte[] data = cache.getCurrentData().getData();

                if (data != null) {
                    ConfigDTO configDTO = new ConfigDTO();
                    configDTO.setBizName(bizName);
                    configDTO.setDataId(dataId);
                    configDTO.setConfig(new String(data, Constants.DEFAULT_CHARSET));
                    listener.receiveConfig(configDTO);
                }

            }
        });

        try {
            cache.start(true);
        } catch (Throwable e) {
            logger.error("Start NodeCache error for path: {}, error info: {}", path, Throwables.getStackTraceAsString(e));
        }

    }

    private void innerRegisterListeners(CuratorFramework zkClient) {
        zkClient.getConnectionStateListenable().addListener(new ConnectionStateListener() {
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                DynConfigClient.logger.info("CuratorFramework state changed: {}", newState);
                if (newState != ConnectionState.CONNECTED && newState == ConnectionState.RECONNECTED) {
                    ;
                }

            }
        });
        zkClient.getUnhandledErrorListenable().addListener(new UnhandledErrorListener() {
            public void unhandledError(String message, Throwable e) {
                DynConfigClient.logger.info("CuratorFramework unhandledError: {}", message);
            }
        });
    }

    /**
     * 设置配置
     *
     * @param path
     * @param value
     * @throws Exception
     */
    public void setConfig(String path, String value) throws Exception {
        if (ZKClient.getClient().checkExists().forPath(path) == null) {
            this.createConfig(path, value);
        } else {
            ZKClient.getClient().setData().forPath(path, value.getBytes(Constants.DEFAULT_CHARSET));
        }

    }

    /**
     * 创建配置
     *
     * @param path
     * @param value
     * @throws Exception
     */
    public void createConfig(String path, String value) throws Exception {
        ZKClient.getClient().create().creatingParentsIfNeeded().forPath(path, value.getBytes(Constants.DEFAULT_CHARSET));
    }

    /**
     * 删除配置
     *
     * @param path
     * @throws Exception
     */
    public void deleteConfig(String path) throws Exception {
        ZKClient.getClient().delete().forPath(path);
    }

    /**
     * 按业务名称+数据项Id获取配置(会组合出一个path=/configs/bizName/dataId)
     *
     * @param bizName
     * @param dataId
     * @return
     * @throws Exception
     */
    public String getConfig(String bizName, String dataId) throws Exception {
        String path = String.format(CONFIG_FORMATTED, new Object[]{bizName, dataId});
        return this.getConfig(path);
    }

    /**
     * 按路径获取配置
     *
     * @param path
     * @return
     * @throws Exception
     */
    public final String getConfig(String path) throws Exception {
        byte[] data = ZKClient.getClient().getData().forPath(path);
        return data == null ? "" : new String(data, Constants.DEFAULT_CHARSET);
    }

    public List<String> getNodes(String path) throws Exception {
        return ZKClient.getClient().getChildren().forPath(path);
    }
}
